package com.bw.gmall.realtime.gongdan3.Day0924;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.gongdan3.Day0923.bean.KeyWordsBean;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class DwdHotWordsTop10 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> pageDs = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_page_log", "xxx").setStartFromEarliest());

//        pageDs.print("页面日志数据------>");

//        SingleOutputStreamOperator<JSONObject> good_listDs = pageDs.flatMap(new FlatMapFunction<String, JSONObject>() {
//            @Override
//            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
//                JSONObject pagejson = JSONObject.parseObject(value);
//
//                if ("good_list".equals(pagejson.getJSONObject("page").getString("page_id"))) {
//                    out.collect(pagejson);
//                }
//            }
//        });

//        SingleOutputStreamOperator<KeyWordsBean> keywords = good_listDs.flatMap(new FlatMapFunction<JSONObject, KeyWordsBean>() {
//            @Override
//            public void flatMap(JSONObject value, Collector<KeyWordsBean> out) throws Exception {
//                KeyWordsBean keyword = new KeyWordsBean();
//
//                if ("keyword".equals(value.getJSONObject("page").getString("item_type"))) {
//                    JSONObject page = value.getJSONObject("page");
//                    String item = page.getString("page_id");
//                    String item_type = page.getString("item_type");
//                    String user_id = value.getJSONObject("common").getString("uid");
//                    keyword.setItem(item);
//                    keyword.setItem_type(item_type);
//                    keyword.setUser_id(user_id);
//                    keyword.setTs(value.getLong("ts"));
//                }
//
//                out.collect(keyword);
//
//            }
//        });

//        keywords.print("关键词--->");


        pageDs.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonDs = JSONObject.parseObject(value);
                JSONObject page = jsonDs.getJSONObject("page");
                if ("search".equals(page.getString("page_id"))){
                    out.collect(jsonDs);
                }
            }
        }).print();



//        // todo 设置水位线
//        SingleOutputStreamOperator<KeyWordsBean> keyword_wms = keywords.assignTimestampsAndWatermarks(WatermarkStrategy.<KeyWordsBean>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<KeyWordsBean>() {
//            @Override
//            public long extractTimestamp(KeyWordsBean element, long recordTimestamp) {
//                return element.getTs();
//            }
//        }));
//        SingleOutputStreamOperator<Tuple4<String, Long, String, String>> aggregate = keyword_wms.keyBy(s -> s.item)
//                .window(TumblingEventTimeWindows.of(Time.minutes(1L)))
//                .aggregate(new AggregateFunction<KeyWordsBean, Tuple2<String, Long>, Tuple2<String, Long>>() {
//                    @Override
//                    public Tuple2<String, Long> createAccumulator() {
//                        return Tuple2.of("", 0L);
//                    }
//
//                    @Override
//                    public Tuple2<String, Long> add(KeyWordsBean value, Tuple2<String, Long> f1) {
//                        f1.f0 = value.item;
//                        f1.f1 += 1L;
//                        return f1;
//                    }
//
//                    @Override
//                    public Tuple2<String, Long> getResult(Tuple2<String, Long> accumulator) {
//                        return accumulator;
//                    }
//
//                    @Override
//                    public Tuple2<String, Long> merge(Tuple2<String, Long> a, Tuple2<String, Long> b) {
//                        return null;
//                    }
//                }, new ProcessWindowFunction<Tuple2<String, Long>, Tuple4<String, Long, String, String>, String, TimeWindow>() {
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        super.open(parameters);
//                    }
//
//                    @Override
//                    public void process(String s, ProcessWindowFunction<Tuple2<String, Long>, Tuple4<String, Long, String, String>, String, TimeWindow>.Context context, Iterable<Tuple2<String, Long>> elements, Collector<Tuple4<String, Long, String, String>> out) throws Exception {
//                        Iterator<Tuple2<String, Long>> iterator = elements.iterator();
//                        while (iterator.hasNext()) {
//                             Tuple2<String, Long> value = iterator.next();
//                            long window_stt = context.window().getStart();
//                            long window_edt = context.window().getEnd();
//                            String stt = DateFormatUtil.toYmdHms(window_stt);
//                            String edt = DateFormatUtil.toYmdHms(window_edt);
//                            out.collect(Tuple4.of(value.f0, value.f1, stt, edt));
//                        }
//                    }
//                });
//
//        aggregate.print("聚合结果--->");






        env.execute();
    }
}
